package com.gzyj.flink.gps.map;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.alibaba.fastjson.support.spring.FastJsonRedisSerializer;
import com.gzyj.flink.driver.Driverlogininfo;
import com.gzyj.flink.dto.alarm.Kafka_VehicleNormalAlarm;
import com.gzyj.flink.gps.GPSPosition;
import com.gzyj.flink.gps.config.RedisKeyPrefixConfig;
import com.gzyj.flink.gps.factory.RedisSessionFactory;
import com.gzyj.flink.gps.map.beidoualarm.BeidouZongDuanAlarm2KafkaVehicleNormalAlarmFun;
import com.gzyj.flink.gps.utils.IntUtils;
import com.gzyj.flink.pojo.gps.BaobiaoAlarmsummary;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.StringRedisSerializer;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Objects;
import java.util.concurrent.TimeUnit;


/**
 * 北斗报警处理
 */
public class BeidouZongDuanAlarmFlow {
    private SingleOutputStreamOperator<GPSPosition> gpsSingleStream;

    public BeidouZongDuanAlarmFlow(SingleOutputStreamOperator<GPSPosition> gpsSingleStream) {
        this.gpsSingleStream = gpsSingleStream;
    }


    public DataStream<BaobiaoAlarmsummary> process() {
        SingleOutputStreamOperator<Kafka_VehicleNormalAlarm> noAlarmStream = this.gpsSingleStream.filter(new FilterFunction<GPSPosition>() {
            @Override
            public boolean filter(GPSPosition gpsPosition) throws Exception {
                return StringUtils.isEmpty(gpsPosition.getAlarmNoteExt());
            }
        }).map(new RichMapFunction<GPSPosition, Kafka_VehicleNormalAlarm>() {

            private LettuceConnectionFactory factory;

            private RedisTemplate<String, Object> getRedisTemplate(LettuceConnectionFactory redisSessionFactory) {
                RedisTemplate<String, Object> redisTemplate = new RedisTemplate<>();
                redisTemplate.setConnectionFactory(redisSessionFactory);
                redisTemplate.setKeySerializer(new StringRedisSerializer());
                redisTemplate.setValueSerializer(new FastJsonRedisSerializer<>(Object.class));
                redisTemplate.setHashKeySerializer(new StringRedisSerializer());
                redisTemplate.setHashValueSerializer(new FastJsonRedisSerializer<>(Object.class));
                redisTemplate.afterPropertiesSet();
                return redisTemplate;
            }


            @Override
            public void open(Configuration parameters) throws Exception {
                factory = RedisSessionFactory.getRedisFactory();
            }

            @Override
            public Kafka_VehicleNormalAlarm map(GPSPosition gpsPosition) throws Exception {
                RedisTemplate<String, Object> clusterRedisTemplate = getRedisTemplate(factory);
                Driverlogininfo driverLoginData = null;

                Object obj = clusterRedisTemplate.opsForValue().get(RedisKeyPrefixConfig.RedisKey_VehicleAlarm + gpsPosition.getVehId());
                Kafka_VehicleNormalAlarm vehicleNormalAlarm = null;
                if (obj == null) {
                    vehicleNormalAlarm = new Kafka_VehicleNormalAlarm();
                } else {
                    vehicleNormalAlarm = JSONObject.parseObject(JSON.toJSONString(obj, SerializerFeature.DisableCircularReferenceDetect), Kafka_VehicleNormalAlarm.class);
                }


                obj = clusterRedisTemplate.opsForValue().get(RedisKeyPrefixConfig.RedisKey_DriverSignOn + gpsPosition.getVehId());

                if (obj == null) {
                    driverLoginData = new Driverlogininfo();
                } else {
                    driverLoginData = JSONObject.parseObject(JSON.toJSONString(obj, SerializerFeature.DisableCircularReferenceDetect), Driverlogininfo.class);
                }
                SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

                Kafka_VehicleNormalAlarm alarm = new Kafka_VehicleNormalAlarm();
                alarm.setVehicleId(gpsPosition.getVehId());
                alarm.setDeptId(gpsPosition.getDeptId());
                alarm.setNote(gpsPosition.getAlarmNoteExt());
                alarm.setVelocity(gpsPosition.getSpeedExt());
                alarm.setLimited(0);
                alarm.setLongitude(gpsPosition.getLngExt());
                alarm.setLatitude(gpsPosition.getLatExt());
                alarm.setElevation(Double.valueOf(gpsPosition.getALTITUDE()));
                alarm.setAngle(gpsPosition.DIRECTION);
                alarm.setLocal(IntUtils.LocalExt(gpsPosition.DIRECTION) == 1);
                alarm.setAnalyzeModel(0);
                alarm.setCreateTime(sdf.format(new Date()));
                alarm.setUpdateTime(sdf.format(new Date()));
                alarm.setTime(gpsPosition.TIME);
                alarm.setIsAlarm(false);
                alarm.setDriver(driverLoginData.getDriverName());
                alarm.setDriverNo(driverLoginData.getDriverNo());
                alarm.setQualificationCode(driverLoginData.getDriverNo());

                clusterRedisTemplate.opsForValue().set(RedisKeyPrefixConfig.RedisKey_VehicleAlarm + gpsPosition.getVehId(), alarm, 8, TimeUnit.HOURS);
                boolean isAlarm = vehicleNormalAlarm.getIsAlarm() == null ? false : vehicleNormalAlarm.getIsAlarm();
                return alarm;
            }
        }).filter(Objects::nonNull);

        BeidouAlarmProcess beidouAlarmProcess = new BeidouAlarmProcess(noAlarmStream);
        DataStream<BaobiaoAlarmsummary> stream1= beidouAlarmProcess.process();

        SingleOutputStreamOperator<Kafka_VehicleNormalAlarm> alarmStream = this.gpsSingleStream.filter(new FilterFunction<GPSPosition>() {
            @Override
            public boolean filter(GPSPosition gpsPosition) throws Exception {
                return StringUtils.isNotEmpty(gpsPosition.getAlarmNoteExt());
            }
        }).map(new BeidouZongDuanAlarm2KafkaVehicleNormalAlarmFun()).filter(Objects::nonNull);

        BeidouAlarmProcess beidouAlarmProcess1 = new BeidouAlarmProcess(alarmStream);
        DataStream<BaobiaoAlarmsummary> stream2= beidouAlarmProcess1.process();

       return stream1.union(stream2);

    }


}
